package com.feiwo.nonego.rpc.consumer.common;

import com.feiwo.nonego.rpc.common.threadpool.ClientThreadPool;
import com.feiwo.nonego.rpc.common.util.IpUtils;
import com.feiwo.nonego.rpc.common.util.NettyUtils;
import com.feiwo.nonego.rpc.common.util.RpcServiceUtils;
import com.feiwo.nonego.rpc.consumer.common.cache.RpcConsumerCache;
import com.feiwo.nonego.rpc.consumer.common.handler.RpcConsumerHandler;
import com.feiwo.nonego.rpc.consumer.common.initializer.RpcConsumerInitializer;
import com.feiwo.nonego.rpc.protocol.RpcProtocol;
import com.feiwo.nonego.rpc.protocol.request.RpcRequest;
import com.feiwo.nonego.rpc.proxy.api.consumer.Consumer;
import com.feiwo.nonego.rpc.proxy.api.future.RpcFuture;
import com.feiwo.nonego.rpc.registry.api.RegistryService;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Objects;
import java.util.Optional;

/**
 *
 *
 * @author feiwo 2025-08-08
 */
public class RpcConsumer implements Consumer {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcConsumer.class);

    private final Bootstrap bootstrap;

    private final EventLoopGroup eventLoopGroup;

    private static  volatile RpcConsumer rpcConsumer;

    private final String localIp;

    public RpcConsumer() {
        localIp = IpUtils.getLocalHostIp();
        bootstrap = new Bootstrap();
        eventLoopGroup = NettyUtils.createEventLoopGroup(4);
        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new RpcConsumerInitializer());
    }

    public static RpcConsumer getInstance() {
        if (rpcConsumer == null) {
            synchronized (RpcConsumer.class) {
                if (rpcConsumer == null) {
                    rpcConsumer = new RpcConsumer();
                }
            }
        }
        return rpcConsumer;
    }

    public void close() {
        RpcConsumerCache.closeRpcClientHandler();
        eventLoopGroup.shutdownGracefully();
        ClientThreadPool.shutdown();
    }

    public Optional<RpcFuture> sendRequest(RpcProtocol<RpcRequest> requestRpcProtocol, RegistryService registryService) throws Exception {
        var request = requestRpcProtocol.getBody();
        var serviceKey = RpcServiceUtils.buildServiceKey(request.getClassName(), request.getVersion(), request.getGroup());
        var parameters = request.getParameters();

        var hashCode = (parameters == null || parameters.length == 0) ? serviceKey.hashCode() : parameters[0].hashCode();
        var serviceMeta = registryService.discovery(serviceKey, hashCode, localIp);
        if (Objects.nonNull(serviceMeta)) {
            var cacheHandler = RpcConsumerCache.get(serviceMeta);
            if (Objects.isNull(cacheHandler)) {
                cacheHandler = getRpcConsumerHandler(serviceMeta.getServiceAddress(), serviceMeta.getServicePort());
                RpcConsumerCache.put(serviceMeta, cacheHandler);
            }else if (!cacheHandler.getChannel().isActive()) {
                cacheHandler.close();
                cacheHandler = getRpcConsumerHandler(serviceMeta.getServiceAddress(), serviceMeta.getServicePort());
                RpcConsumerCache.put(serviceMeta, cacheHandler);
            }
            return cacheHandler.sendRequest(requestRpcProtocol, requestRpcProtocol.getBody().isAsync(), requestRpcProtocol.getBody().isOneway());
        }
        return Optional.empty();
    }

    private RpcConsumerHandler getRpcConsumerHandler(String serviceAddress, int port) throws InterruptedException {
        var channelFuture = bootstrap.connect(serviceAddress,port).sync();
        channelFuture.addListener(future -> {
            if (future.isSuccess()) {
                LOGGER.info("connect rpc server {} on port {} success", serviceAddress, port);
            }else {
                LOGGER.error("connect rpc server {} on port {} failed", serviceAddress, port);
                LOGGER.error("channel future ex ---> {}", channelFuture.cause().getMessage());
                eventLoopGroup.shutdownGracefully();
            }
        });
        return channelFuture.channel().pipeline().get(RpcConsumerHandler.class);
    }
}
